package com.abyss.watermark;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 对长期延迟数据的处理
 */
public class LatenessDataDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 标记为使用事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // 设置全局并行度为1
        env.setParallelism(1);

        // Source
        DataStreamSource<String> socketTextStream = env.socketTextStream("node1", 9999);

        // 将输入的[单词 时间戳]字符串, 转换为元组
        SingleOutputStreamOperator<Tuple2<String, Long>> tuple = socketTextStream.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String value) throws Exception {
                String[] ele = value.split(" ");
                return Tuple2.of(ele[0], Long.parseLong(ele[1]));
            }
        });

        // 设置事件时间和watermark
        SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = tuple
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<String, Long>>(Time.seconds(3)) {
                    @Override
                    public long extractTimestamp(Tuple2<String, Long> element) {
                        // 这里打印watermark不准确会滞后, 因为我们复写的extractTimestamp还有一个重载的方法
                        // public final long extractTimestamp(T element, long previousElementTimestamp)
                        // 在这个双参数的方法里才会更新currentMaxTimestamp这个变量
                        // 但是这个方法是final 的无法复写, 所以我们如果使用BoundedOutOfOrdernessTimestampExtractor类
                        // 就基本上无法手动打印出准确的watermark
                        // 只能自行实现AssignerWithPeriodicWatermarks
//                System.out.println("Event time: " + element.f1 + ", " +
//                        "watermark: " + getCurrentWatermark().getTimestamp());
                        return element.f1;
                    }
                });

        // 配置窗口, 滚动窗口, 长度5秒
        WindowedStream<Tuple2<String, Long>, Tuple, TimeWindow> windowedStream = watermarks.keyBy(0).timeWindow(Time.seconds(5));

        // 增加窗口对长期延迟数据的处理
        WindowedStream<Tuple2<String, Long>, Tuple, TimeWindow> windowedStreamWithLateness = windowedStream
                .allowedLateness(Time.seconds(2));

        // 计算分组后每隔单词在窗口内出现的个数
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = windowedStreamWithLateness.apply(new WindowFunction<Tuple2<String, Long>, Tuple2<String, Integer>, Tuple, TimeWindow>() {
            @Override
            public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<Tuple2<String, Integer>> out) throws Exception {
                String key = null;
                int count = 0;
                for (Tuple2<String, Long> ele : input) {
                    count += 1;
                    key = ele.f0;
                }
                out.collect(Tuple2.of(key, count));
                System.out.println("window start: " + window.getStart() + ", " +
                        "window end: " + window.getEnd());
            }
        });

        result.print();

        env.execute();
    }
}